{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "bd5e22a6-30bc-4ef8-a1d0-cb79f679cd78",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "83949cc5-dcca-4cc8-b5e3-43c6ff6ff00c",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder.master(\"local[*]\").appName(\"rdd_testing\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "52c7b7e0-d799-4d55-a88d-6412efb38d15",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[0] at readRDDFromFile at PythonRDD.scala:289"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "collection = [1,\"two\",3.0,(\"four\",4),{\"five\":5}]\n",
    "sc = spark.sparkContext\n",
    "collection_rdd = sc.parallelize(collection)\n",
    "collection_rdd "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "588b7f4b-bb1c-4c29-9216-15c26acb6053",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[2, 'two', 4.0, ('four', 4), {'five': 5}]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def safer_add_one(value):\n",
    "    try:\n",
    "        return value+1\n",
    "    except TypeError:\n",
    "        return value\n",
    "\n",
    "collection_rdd.map(safer_add_one).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "5d318d39-7663-4e78-9248-ddbb9bdfa731",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1, 3.0]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "collection_rdd.filter(lambda x:isinstance(x,(int,float))).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "01412198-38f1-40a6-be03-5e52b4e7fceb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "24"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from operator import add \n",
    "\n",
    "collection_rdd = sc.parallelize([4,7,9,1,3])\n",
    "collection_rdd.reduce(add)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5866a5f0-3e52-4838-8528-d63ef93addd1",
   "metadata": {},
   "source": [
    "exercise 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "1c0d0865-57ea-4c1e-8ce0-693e945b2446",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "collection_rdd = sc.parallelize([4,7,7,9,1,1,1,3])\n",
    "collection_rdd.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "2aba046f-8de9-4f1e-9358-5283e6b1b5da",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "collection_rdd.map(lambda x:1).reduce(add)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "334656fa-2cff-4ee5-a1db-2df9599ff35c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql.types as T"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "c915f256-bc7f-4328-8d8f-2955d4e9361d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----------+\n",
      "|numerator|denominator|\n",
      "+---------+-----------+\n",
      "|        0|          1|\n",
      "|        0|          2|\n",
      "|        0|          3|\n",
      "|        0|          4|\n",
      "|        1|          1|\n",
      "+---------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fractions = [[x,y] for x in range(5) for y in range(1,5)]\n",
    "frac_df = spark.createDataFrame(fractions,[\"numerator\",\"denominator\"])\n",
    "frac_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "bfe2c5e8-0ae4-4ac7-97ab-e446508318a3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|fraction|\n",
      "+--------+\n",
      "|[0, 1]  |\n",
      "|[0, 2]  |\n",
      "|[0, 3]  |\n",
      "|[0, 4]  |\n",
      "|[1, 1]  |\n",
      "+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "frac_df = frac_df.select(F.array(F.col(\"numerator\"),F.col(\"denominator\")).alias(\"fraction\"))\n",
    "frac_df.show(5,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "0b927eb5-7330-45a3-ad10-504d9cbabe57",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fractions import Fraction\n",
    "from typing import Tuple,Optional\n",
    "\n",
    "Frac = Tuple[int,int]\n",
    "\n",
    "def py_reduce_fraction(frac:Frac) -> Optional[Frac]:\n",
    "    num,denom = frac\n",
    "    if denom:\n",
    "        answer = Fraction(num,denom)\n",
    "        return answer.numerator, answer.denominator\n",
    "    return None\n",
    "\n",
    "def py_fraction_to_float(frac:Frac) -> Optional[float]:\n",
    "    num,denom = frac\n",
    "    if denom:\n",
    "        return num/denom\n",
    "    return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "ce09cf53-3448-4636-a9f9-936ba9bedaa0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----------------+--------------+\n",
      "|fraction|reduced_fraction|fraction_float|\n",
      "+--------+----------------+--------------+\n",
      "|  [0, 1]|          [0, 1]|        [0, 1]|\n",
      "|  [0, 2]|          [0, 1]|        [0, 1]|\n",
      "|  [0, 3]|          [0, 1]|        [0, 1]|\n",
      "|  [0, 4]|          [0, 1]|        [0, 1]|\n",
      "|  [1, 1]|          [1, 1]|        [2, 1]|\n",
      "|  [1, 2]|          [1, 2]|        [1, 1]|\n",
      "|  [1, 3]|          [1, 3]|        [2, 3]|\n",
      "|  [1, 4]|          [1, 4]|        [1, 2]|\n",
      "|  [2, 1]|          [2, 1]|        [4, 1]|\n",
      "|  [2, 2]|          [1, 1]|        [2, 1]|\n",
      "+--------+----------------+--------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "SparkFrac = T.ArrayType(T.LongType())\n",
    "\n",
    "reduce_function = F.udf(py_reduce_fraction, SparkFrac)\n",
    "frac_df = frac_df.withColumn(\"reduced_fraction\",reduce_function(\"fraction\"))\n",
    "frac_df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "f5fe18a9-820d-473e-b7fb-bb8b2f9e41c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "@F.udf(T.DoubleType())\n",
    "def py_fraction_to_float(frac:Frac) -> Optional[float]:\n",
    "    num,denom = frac\n",
    "    if denom:\n",
    "        return num/denom\n",
    "    return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "21baf3d2-d1a1-41f5-8fb9-77506f1ebecd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----------------+------------------+\n",
      "|fraction|reduced_fraction|    fraction_float|\n",
      "+--------+----------------+------------------+\n",
      "|  [0, 1]|          [0, 1]|               0.0|\n",
      "|  [0, 2]|          [0, 1]|               0.0|\n",
      "|  [0, 3]|          [0, 1]|               0.0|\n",
      "|  [0, 4]|          [0, 1]|               0.0|\n",
      "|  [1, 1]|          [1, 1]|               1.0|\n",
      "|  [1, 2]|          [1, 2]|               0.5|\n",
      "|  [1, 3]|          [1, 3]|0.3333333333333333|\n",
      "|  [1, 4]|          [1, 4]|              0.25|\n",
      "|  [2, 1]|          [2, 1]|               2.0|\n",
      "|  [2, 2]|          [1, 1]|               1.0|\n",
      "+--------+----------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "frac_df = frac_df.withColumn(\"fraction_float\",py_fraction_to_float(\"fraction\"))\n",
    "frac_df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "34bd0443-a04c-4f28-b28e-2fea171256aa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----------------+------------------+-----------------+\n",
      "|fraction|reduced_fraction|    fraction_float|fraction_addition|\n",
      "+--------+----------------+------------------+-----------------+\n",
      "|  [0, 1]|          [0, 1]|               0.0|           [0, 1]|\n",
      "|  [0, 2]|          [0, 1]|               0.0|           [0, 1]|\n",
      "|  [0, 3]|          [0, 1]|               0.0|           [0, 1]|\n",
      "|  [0, 4]|          [0, 1]|               0.0|           [0, 1]|\n",
      "|  [1, 1]|          [1, 1]|               1.0|           [2, 1]|\n",
      "|  [1, 2]|          [1, 2]|               0.5|           [1, 1]|\n",
      "|  [1, 3]|          [1, 3]|0.3333333333333333|           [2, 3]|\n",
      "|  [1, 4]|          [1, 4]|              0.25|           [1, 2]|\n",
      "|  [2, 1]|          [2, 1]|               2.0|           [4, 1]|\n",
      "|  [2, 2]|          [1, 1]|               1.0|           [2, 1]|\n",
      "+--------+----------------+------------------+-----------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@F.udf(T.ArrayType(T.LongType()))\n",
    "def fraction_addition(left:Frac,right:Frac)->Optional[Frac]:\n",
    "    left_fraction = Fraction(*left)\n",
    "    right_fraction= Fraction(*right)\n",
    "    res = left_fraction+right_fraction\n",
    "    return res.numerator, res.denominator\n",
    "\n",
    "frac_df = frac_df.withColumn(\"fraction_addition\",fraction_addition(\"reduced_fraction\",\"reduced_fraction\"))\n",
    "frac_df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "28bdfedd-a0ee-4695-9059-90a41d30b8ca",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
